package com.navinfo.opentsp.platform.computing.analysis.entity;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.navinfo.opentsp.platform.computing.analysis.util.BigDataUtils;
import com.navinfo.opentsp.platform.computing.analysis.util.MongoDBUtil;
import com.navinfo.opentsp.platform.computing.analysis.util.MysqlUtil;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.*;
import org.apache.spark.sql.catalyst.expressions.GenericRow;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.JavaConverters;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

@Data
public class VehicleDrivingNumberInGridDto implements Serializable {
    
    private static final Logger logger = LoggerFactory.getLogger(VehicleDrivingNumberInGridDto.class);

    //构建schema嵌套层数
    private static final int num = 3;

    /**
     * 地区编码
     */
    private long district;
    /**
     * 地区名称
     */
    private String name;
    /**
     * 瓦片ID
     */
    private long tile;
    /**
     * 经过地区次数
     */
    private int times;
    /**
     * 时间月份
     */
    private int month;
    /**
     * 车辆明细信息
     */
    private List<HashMap> detail;

    public enum DataLoadTypeEnum{
        MYSQL{
            @Override
            public Object createRow(VehicleDrivingNumberInGridDto vehicleDrivingNumberInGridDto, String day) {
                List<Object> list = new ArrayList<>();
                list.add(vehicleDrivingNumberInGridDto.getDistrict());
                list.add(vehicleDrivingNumberInGridDto.getName());
                list.add(vehicleDrivingNumberInGridDto.getTile());
                list.add(vehicleDrivingNumberInGridDto.getTimes());
                list.add(vehicleDrivingNumberInGridDto.getMonth());
                List<HashMap> detail = vehicleDrivingNumberInGridDto.getDetail();
                String detailRow = JSON.toJSONString(detail);
                list.add(detailRow);
                list.add(Integer.parseInt(day));
                return new GenericRow(list.toArray());
            }

            @Override
            public StructType createSchema() {
                List<StructField> schema = new ArrayList<>();
                schema.add(new StructField("district", DataTypes.LongType,true, Metadata.empty()));
                schema.add(new StructField("name",DataTypes.StringType,true,Metadata.empty()));
                schema.add(new StructField("tile", DataTypes.LongType,true, Metadata.empty()));
                schema.add(new StructField("times",DataTypes.IntegerType,true,Metadata.empty()));
                schema.add(new StructField("month",DataTypes.IntegerType,true,Metadata.empty()));
                schema.add(new StructField("detail", DataTypes.StringType,true,Metadata.empty()));
                schema.add(new StructField("day",DataTypes.IntegerType,true,Metadata.empty()));
                return new StructType(schema.toArray(new StructField[schema.size()]));
            }

            @Override
            public void save(SparkSession spark, JavaRDD rows, String day) {


                String url = PropertiesUtil.getProperties("mysql.save.vehicle.jdbc.url");
                String user = PropertiesUtil.getProperties("mysql.save.vehicle.user");
                String passwd = PropertiesUtil.getProperties("mysql.save.vehicle.password");
                String srcdbtable = PropertiesUtil.getProperties("mysql.save.vehicleDrivingNumber.dbtable");
                int connection = Integer.parseInt(Optional.ofNullable(PropertiesUtil.getProperties("mysql.save.vehicle.connection")).orElse("1"));


                //取分表表名
                String dbtable = srcdbtable+"_"+ StringUtils.substring(day,2,6);
                clear(url,user,passwd,dbtable,day);

                Dataset<Row> VehicleDrivingNumberInGridDs = spark.createDataFrame(rows,createSchema());
                logger.error("插入{}表记录数:{}",dbtable,VehicleDrivingNumberInGridDs.count());
                VehicleDrivingNumberInGridDs.repartition(connection);

                VehicleDrivingNumberInGridDs.write().mode(SaveMode.Append).format("jdbc")
                        .option("driver","com.mysql.cj.jdbc.Driver")
                        .option("url",url)
                        .option("dbtable",dbtable)
                        .option("user",user)
                        .option("password",passwd)
                        .save();

            }

            private void clear(String url,String user,String password,String tbname,String day){

                logger.error("开始清除当日瓦片车次数据");
                Connection conn = null;
                try{
                    conn = MysqlUtil.getConnection(url,user,password);
                    String sql = "delete from "+tbname+" where day=?";
                    PreparedStatement prep = conn.prepareStatement(sql);
                    //prep.setString(1,tbname);
                    prep.setInt(1,Integer.parseInt(day));
                    int rs = prep.executeUpdate();
                    logger.info("{}表{}日瓦片车次数据删除{}条记录",tbname,day,rs);

                }catch (SQLException e){
                    logger.error("瓦片统计数据库连接失败:{}",e);
                }catch (Exception e){
                    logger.error("删除当日瓦片车次统计数据失败:{}",e);
                }finally {
                    MysqlUtil.closeConnection(conn);
                }

            }
        },
        HIVE{
            @Override
            public Object createRow(VehicleDrivingNumberInGridDto vehicleDrivingNumberInGridDto,String day) {
                List<Object> list = new ArrayList<>();
                list.add(vehicleDrivingNumberInGridDto.getDistrict());
                list.add(vehicleDrivingNumberInGridDto.getName());
                list.add(vehicleDrivingNumberInGridDto.getTile());
                list.add(vehicleDrivingNumberInGridDto.getTimes());
                list.add(vehicleDrivingNumberInGridDto.getMonth());
                List<HashMap> detail = vehicleDrivingNumberInGridDto.getDetail();
                scala.collection.Seq<Row> detailRow = detailToRow(num,detail);
                list.add(detailRow);
                return new GenericRow(list.toArray());
            }

            @Override
            public StructType createSchema() {
                List<StructField> schema = new ArrayList<>();
                schema.add(new StructField("district", DataTypes.LongType,true, Metadata.empty()));
                schema.add(new StructField("name",DataTypes.StringType,true,Metadata.empty()));
                schema.add(new StructField("tile", DataTypes.LongType,true, Metadata.empty()));
                schema.add(new StructField("times",DataTypes.IntegerType,true,Metadata.empty()));
                schema.add(new StructField("month",DataTypes.IntegerType,true,Metadata.empty()));
                //递归生成符合嵌套类型
                List<StructField> list = getDetailList(num);
                schema.add(new StructField("detail", DataTypes.createArrayType(DataTypes.createStructType(list)),true,Metadata.empty()));
                return new StructType(schema.toArray(new StructField[schema.size()]));
            }

            @Override
            public void save(SparkSession spark, JavaRDD rows, String day) {
                Dataset<Row> VehicleDrivingNumberInGridDs = spark.createDataFrame(rows,createSchema());
                BigDataUtils.saveHiveTable(spark,VehicleDrivingNumberInGridDs,
                        BigDataUtils.CoalesceType.valueOf(PropertiesUtil.getProperties("hive.save.vehicleDrivingNumber.coalesceType")),
                        Integer.parseInt(PropertiesUtil.getProperties("hive.save.vehicleDrivingNumber.coalesceNumber")),
                        PropertiesUtil.getProperties("hive.save.vehicleDrivingNumber.db"),
                        PropertiesUtil.getProperties("hive.save.vehicleDrivingNumber.table"),
                        Lists.newArrayList("day"),
                        Lists.newArrayList(day),
                        Arrays.asList(createSchema().fieldNames())
                );
            }
        },
        MONGO{
            @Override
            public Object createRow(VehicleDrivingNumberInGridDto vehicleDrivingNumberInGridDto, String day) {
                Document doc = new Document();
                try {
                    Map map = new HashMap();
                    map.put("district",vehicleDrivingNumberInGridDto.getDistrict());
                    map.put("name",vehicleDrivingNumberInGridDto.getName());
                    map.put("tile", vehicleDrivingNumberInGridDto.getTile());
                    map.put("times", vehicleDrivingNumberInGridDto.getTimes());
                    map.put("month", vehicleDrivingNumberInGridDto.getMonth());
                    map.put("dl", vehicleDrivingNumberInGridDto.getDetail());
                    doc.putAll(map);
                } catch (Exception e) {
                    //logger.error("Document parse error", e);
                }
                return doc;
            }

            @Override
            public StructType createSchema() {
                return null;
            }

            @Override
            public void save(SparkSession spark, JavaRDD rows, String day) {
                //组装Mongo集合名称
                String tablename = PropertiesUtil.getProperties("mongo.save.vehicleDrivingNumber.table")+ "_" + day.substring(2, 6);
                //清空当日数据
                //瓦片统计没有day字段，不能清空当日数据
                //clear(tablename,day);
                //重分区
                int repartitionNum = Integer.parseInt(PropertiesUtil.getProperties("mongo.save.vehicle.coalesceNumber"));
                rows.repartition(repartitionNum);
                //存入MongoDB
                MongoDBUtil.saveByJavaRDD(JavaSparkContext.fromSparkContext(spark.sparkContext()),rows,tablename);
            }

            public void clear(String collection,String day){
                String url = PropertiesUtil.getProperties("mongo.clear.vehicle.url");
                String connectionsPerHost = PropertiesUtil.getProperties("mongo.clear.vehicle.connectionsPerHost");
                String db = PropertiesUtil.getProperties("mongo.clear.vehicle.db");
                String username = PropertiesUtil.getProperties("mongo.clear.vehicle.username");
                String password = PropertiesUtil.getProperties("mongo.clear.vehicle.password");
                HashMap<String, String> Map = new HashMap<>();
                Map.put("mongo.url", url);
                Map.put("mongo.connectionsPerHost", connectionsPerHost);
                Map.put("mongo.database", db);
                Map.put("mongo.collection", collection);
                Map.put("mongo.username", username);
                Map.put("mongo.password", password);
                try {
                    MongoDBUtil.buildInstance(Map);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                MongoDBUtil mongo = MongoDBUtil.getInstance();
                long result = mongo.deleteByDay(mongo.getCollection(db,collection),Integer.parseInt(day));
                logger.error("清空瓦片统计Mongo数据，删除日期:{}，记录数:{}",day,result);
                mongo.close();
            }
        };

        DataLoadTypeEnum(){}
        public abstract Object createRow(VehicleDrivingNumberInGridDto vehicleDrivingNumberInGridDto,String day);
        public abstract StructType createSchema();
        public abstract void save(SparkSession spark, JavaRDD rows , String day );
    }

    private static scala.collection.Seq<Row> detailToRow (int num ,List<HashMap> detailList){
        --num;//递归次数，为0跳出
        if (detailList == null || detailList.isEmpty()) {
            return null;
        }
        if(num == 0){
            List<Row> lastList = new ArrayList<>();
            for(HashMap map1:detailList){
                int d = Integer.parseInt(String.valueOf(Optional.ofNullable(map1.get("d")).orElse("0")));
                long tid = Long.parseLong(String.valueOf(Optional.ofNullable(map1.get("tid")).orElse("0")));
                //List<HashMap> list = castList(map1.get("dl"),HashMap.class);
                Row detailrow = RowFactory.create(
                        d,tid
                        //detailToRow(num,list)
                );
                lastList.add(detailrow);
            }
            return JavaConverters.asScalaIteratorConverter(lastList.iterator()).asScala().toSeq();
        }
        List<Row> rowList = new ArrayList<>();
        //处理第一层
        for(Map map1:detailList){
            long tile = Long.parseLong(String.valueOf(map1.get("tile")));
            int times = Integer.parseInt(String.valueOf(map1.get("times")));
            Row detailrow = null;
            //if(times > 0) {
                //obj 转换成原来的 List<Map>
                List<HashMap> list = castList(map1.get("dl"), HashMap.class);
                detailrow = RowFactory.create(
                        tile, times,
                        detailToRow(num, list)
                );
            rowList.add(detailrow);
        }
        return JavaConverters.asScalaIteratorConverter(rowList.iterator()).asScala().toSeq();
}


    private static List<StructField> getDetailList (int num){

        List<StructField> list = new ArrayList<>();
        //递归次数
        --num;
        if(num == 0){
            list.add(new StructField("d",DataTypes.IntegerType,true,Metadata.empty()));
            list.add(new StructField("tid",DataTypes.LongType,true,Metadata.empty()));
            return list;
        }
        list.add(new StructField("tile",DataTypes.LongType,true,Metadata.empty()));
        list.add(new StructField("times",DataTypes.IntegerType,true,Metadata.empty()));
        list.add(new StructField("dl",DataTypes.createArrayType(DataTypes.createStructType(getDetailList(num))),true,Metadata.empty()));
        return list;
    }

    public static <T> List<T> castList(Object obj, Class<T> clazz)
    {
        List<T> result = new ArrayList<T>();
        //HashMap map = new HashMap();
        if(obj instanceof HashSet<?>){
            //obj = ((HashSet) obj).iterator().next();
            HashSet<HashMap> set = (HashSet) obj;
            for (Object o : set){
                result.add(clazz.cast(o));
            }
            return result;
        }
        if(obj instanceof List<?>)
        {
            for (Object o : (List<?>) obj)
            {
                result.add(clazz.cast(o));
            }
            return result;
        }
        return null;
    }






}
